package br.unicamp.mc715.transport;

import java.io.IOException;
import java.io.Serializable;
import java.net.BindException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MulticastSocket;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.Queue;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import br.unicamp.mc715.common.MC715IOException;
import br.unicamp.mc715.common.MC715SerializationException;
import br.unicamp.mc715.common.Marshall;
import br.unicamp.mc715.common.RandomString;
import br.unicamp.mc715.order.PingMessage;


/*
 * Essa implementação foi feita de boa-vontade, a partir dos códigos que estavam no controle de versão.
 * Copiei as classes que estavam na revisão Zero e apaguei as implementações sem olhar para elas.
 */
/**
 * This class implements a UDP multicast transport. It allows to send and
 * receive multicast and unicast messages. Unicast addressing is done using
 * the transport id object returned by <code>getId()</code>. <p>
 * 
 * This class needs a functional UDP multicast route to work. Multicast routing
 * is no different than unicast routing: packets are routed accordingly to the
 * routes netmasks and the multicast IP. As it is very uncommon to have explicit
 * multicast routes, multicast packets will be usually routed to the default
 * gateway subnet. For hosts with only one interface, it is necessary that this
 * subnet delivers multicast packets correctly (not firewalled). If the host has
 * two or more interfaces and the default gateway subnet isn't suitable for
 * multicast, an explicit route for multicast traffic must be created pointing
 * to a suitable subnet. For example, in Linux a route could be set with this
 * command: <p>
 * <code>   route add -net 224.0.0.0/4 eth0</code> <p>
 * On windows use:
 * 			route -p ADD 224.0.0.0 MASK 240.0.0.0 192.168.0.1
 */

public class UDPTransport implements Transport {

	public static final UDPTransportId multicastEchoId      = new UDPTransportId(new byte[]{(byte)224,0,0,0}, 5678);
	public static final UDPTransportId multicastTransportId = new UDPTransportId(new byte[]{(byte)224,0,0,0}, 5679);
	UDPTransportId myId;
	DatagramSocket myDatagramSocket;
	MulticastSocket multicastSocket;

	protected static final int MAX_DATA_SIZE = 65507; /* 20 bytes IP, 8 bytes UDP */
	protected static final int HEADER_SIZE = 6; 
	/**Fila de mensagens aguardando tratamento*/
	protected Queue<UDPMessage> messageQueue = new LinkedList<UDPMessage>();

	/**Este semaforo controla a espera de mensagens na fila */
	Semaphore queueSemaphore = new Semaphore(0);

	/**
	 * Creates a new UDP transport with a default multicast IP and port. <p>
	 * 
	 * @throws MC715IOException if an I/O error occurs.
	 */
	public UDPTransport() throws MC715IOException {
		try {
			// Cria o socket unicast
			InetAddress addr = allocAddress();
			int port = 4673;
			while (true) {
				try {
					myDatagramSocket = new DatagramSocket(port, addr);
				} catch (BindException e) {
					port++;
					continue;
				}
				break;
			}
			myId = new UDPTransportId((InetSocketAddress) myDatagramSocket.getLocalSocketAddress());

			// Cria o socket multicast
			multicastSocket = new MulticastSocket(multicastTransportId.getSocketAddress().getPort());
			multicastSocket.joinGroup(multicastTransportId.getSocketAddress().getAddress());

			// Começa a receber as mensagens em outra thread.
			new ReceiveMessageThread(myDatagramSocket).start();
			new ReceiveMessageThread(multicastSocket).start();

		} catch (IOException e) {
			throw new MC715IOException(e);
		}
	}

	private class ReceiveMessageThread extends Thread {

		private final DatagramSocket socket;

		public ReceiveMessageThread(DatagramSocket socket) {
			super("UDP-Receive-" + socket.getLocalSocketAddress());
			setDaemon(true);
			this.socket = socket;
		}

		public void run() {
			byte[] buf = new byte[MAX_DATA_SIZE];
			DatagramPacket pack = new DatagramPacket(buf, buf.length);

			while (true) {
				try {
					socket.receive(pack);
				} catch (Exception e) {
					throw new RuntimeException(e);
				}

				try {
					Serializable payload = Marshall.unmarshall(buf);

					/* Se o payload eh ping message adiciono timestamp */
					if (payload instanceof PingMessage) {
						((PingMessage)payload).append(System.nanoTime());
						//Log.logger.info("recv="+((PingMessage)payload).toString());
					}

					UDPTransportId sender = new UDPTransportId((InetSocketAddress)pack.getSocketAddress());
					UDPMessage message = new UDPMessage(payload, sender, myId);
					messageReceived(message);
				} catch (Exception e) {
					System.err.println("Invalid message from: " + pack.getSocketAddress());
				}
			}
		}
	}

	public TransportId getId() { 
		return myId;
	}

	private void messageReceived(UDPMessage message) {
		synchronized (messageQueue) {
			messageQueue.add(message);
		}
		queueSemaphore.release();
	}

	public UDPMessage receiveMessage() {
		try {
			queueSemaphore.acquire();
		} catch (InterruptedException e) {
			return null;
		}

		synchronized (messageQueue) {
			return messageQueue.remove();
		}
	}

	public UDPMessage receiveMessage(int timeout) {
		try {
			boolean acquired = queueSemaphore.tryAcquire(timeout, TimeUnit.MILLISECONDS);
			if (!acquired)
				return null;
		} catch (InterruptedException e) {
			return null;
		}

		synchronized (messageQueue) {
			return messageQueue.remove();
		}
	}

	public void sendMessage(Serializable message) throws MC715IOException, MC715SerializationException {
		sendMessage(message, multicastTransportId);
	}

	public void sendMessage(Serializable message, TransportId destination) throws MC715IOException, MC715SerializationException {

		/* Se o payload eh ping message adiciono timestamp */
		if (message instanceof PingMessage) {
			PingMessage payload = (PingMessage)message;
			payload.append(System.nanoTime());
		}

		byte[] buf = Marshall.marshall(message);
		try {
			DatagramPacket pack = new DatagramPacket(buf, buf.length, ((UDPTransportId)destination).getSocketAddress());
			myDatagramSocket.send(pack);
		} catch (IOException e) {
			throw new MC715IOException(e);
		}
	}

	/** Acha o endereço que deve ser usado no bind() */
	private InetAddress allocAddress() throws MC715IOException{
		try {
			//Envia um pacote com uma assinatura gerada na hora. Isso garante que o pacote recebido foi gerado por este transporte
			byte[] signature = RandomString.RANDOM_256.nextString().getBytes();
			final DatagramPacket signaturePacket = new DatagramPacket(signature, signature.length, multicastEchoId.getSocketAddress());

			final MulticastSocket echoSocket = new MulticastSocket(multicastEchoId.getSocketAddress().getPort());
			echoSocket.joinGroup(multicastEchoId.getSocketAddress().getAddress());

			Timer timer = new Timer("Multicast address search");
			timer.schedule(new TimerTask() {
				@Override
				public void run() {
					try {
						echoSocket.send(signaturePacket);
					} catch (IOException e) {
						throw new RuntimeException(e);
					}
				}
			}, 0, 100);


			DatagramPacket recvPacket = new DatagramPacket(new byte[256], 256);
			while (true) {
				echoSocket.receive(recvPacket);
				if (recvPacket.getLength() != signature.length) continue;
				if (!Arrays.equals(signature, recvPacket.getData())) continue;
				break; //Mensagem recebida de mim mesmo -> Devo usar o endereço IP remetente como meu ID
			}
			timer.cancel();
			echoSocket.close();
			return recvPacket.getAddress();
		} catch (IOException e) {
			throw new MC715IOException(e);
		}
	}
	
	public void clearBuffer()
	{
		synchronized (messageQueue) {
			messageQueue.clear();
		}
		queueSemaphore.drainPermits();
	}
}
